package cn.llin.kafka;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;


import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.UUID;

@Service
public class MhdsjzxApplication {

    private static final String TOPIC="topic_test_xh";
    private static final String CONFIG_DIR="D:\\MT\\赣州\\视频云对接20230217\\星环kafka对接使用说明";
    private static final String BOOTSTRAP_SERVERS="11.100.5.102:32610,11.100.5.101:32283,11.100.5.99:31994";
    private static final String TOS_PRINCIPAL="tos_k2142rz";


    public void Mhdsjzx() throws InterruptedException {


        System.setProperty("java.security.auth.login.config",CONFIG_DIR + "jaas.conf");
        System.setProperty("java.security.krb5.conf",CONFIG_DIR + "krb5.conf");

        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        // 集群kafka bootstrap-server
        props.put("security.protocol", "SASL_PLAINTEXT");// security.protocol
        props.put("sasl.mechanism", "GSSAPI");// sasl.mechanism
        props.put("sasl.kerberos.service.name", "kafka");// sasl.kerberos.service.name
        props.put("sasl.kerberos.service.principal.instance", TOS_PRINCIPAL);
        props.put(ProducerConfig.ACKS_CONFIG, "all");// acks
        props.put(ProducerConfig.RETRIES_CONFIG, 3);// retries

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        SimpleDateFormat sdf =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SSS" );

        for (int i = 10; i < 20; i++){

            String s = i + "|" + sdf.format(new Date()) + "|" + UUID.randomUUID().toString();
            System.out.println(s);
            try{
//                producer.send(new ProducerRecord<>(TOPIC,s ));
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("exception.........");
            } finally {
                Thread.sleep(1000);
            }
        }
    }
}
